package com.jacken.mars.commons.mq.base.mq.producer.impl;

import com.jacken.mars.commons.enums.MQEnum;
import com.jacken.mars.commons.mq.base.mq.producer.MqProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author WQ
 * @version 1.0
 * @date 2021/1/4 9:38
 */
@Slf4j
@Component
public class Test01MqProducer implements MqProducer {


    private DefaultMQProducer producer;

    @Value("${rocketmq.nameServer}")
    private String nameServer;

    //@PostConstruct
    @Override
    public void init() {
        //示例生产者
        producer = new DefaultMQProducer(MQEnum.TEST_01_MQ.getGroup());
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        //绑定name server
        producer.setNamesrvAddr(nameServer);
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
            log.info("Producer 启动失败");
        }
    }


    /**
     * 发送消息
     *
     * @param topic   主题
     * @param message 消息
     */
    public void send(String topic, String message) {
        try {
            producer.send(new Message(topic, message.getBytes()));
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
            log.error(topic + "===>" + message + "发送失败");
        }
    }

    /**
     * 发送消息
     *
     * @param topic   主体
     * @param tag     tag
     * @param message 消息内容
     */
    public void send(String topic, String tag, String message) {
        try {
            producer.send(new Message(topic, tag, message.getBytes()));
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
            log.error(topic + "===>" + message + "发送失败");
        }
    }

    /**
     * 发送消息
     *
     * @param topic   主题
     * @param tag     tag
     * @param keys    keys
     * @param message 消息
     */
    public void send(String topic, String tag, String keys, String message) {
        try {
            producer.send(new Message(topic, tag, keys, message.getBytes()));
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
            log.error(topic + "===>" + message + "发送失败");
        }
    }

    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }


}
